package avicit.bdp.dcs.task.service;

import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.IotDbSessionUtils;
import avicit.bdp.common.utils.enums.FileType;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.TableService;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.json.service.DataxJsonService;
import avicit.bdp.dcs.semistructure.dto.HiveJsonDto;
import avicit.bdp.dcs.semistructure.dto.IotdbJsonDto;
import avicit.bdp.dcs.semistructure.service.TemplateFieldService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.ProcessMode;
import avicit.bdp.dcs.tools.enums.SyncCondition;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.tools.enums.WriteMode;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.JdbcConstants;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2023-04-07 16:12
 * @类说明： 文件采集任务服务类
 * @修改记录：
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class FileTaskService implements TaskInterface {

  private static final Logger logger = LoggerFactory.getLogger(FileTaskService.class);

  @Autowired private TaskService taskService;
  @Autowired private DcsJobService jobService;
  @Autowired private TemplateFieldService templateFieldService;
  @Autowired private DataxJsonService dataxJsonService;
  @Autowired private TableService tableService;
  @Autowired private TaskTableDetailService taskTableDetailService;
  @Autowired private TaskDataDetailService taskDataDetailService;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BaseDbService baseDbService;

  /** 查看任务 */
  @Override
  public TaskDTO getTaskById(JobDTO jobDTO, String srcTableName) throws Exception {
    // 1、查询任务
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setSrcTableName(srcTableName);
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO task = taskService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        task, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobDTO.getId()));
    task.setDataxJson(null);

    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());

    String iotdbJson = task.getIotdbJson();
    if (StringUtils.isNotEmpty(iotdbJson)) {
      IotdbJsonDto iotdbJsonDto = JSON.parseObject(iotdbJson, IotdbJsonDto.class);
      task.setIotdbJsonDto(iotdbJsonDto);
    }

    String hiveJson = task.getHiveJson();
    if (StringUtils.isNotEmpty(hiveJson)) {
      HiveJsonDto hiveJsonDto = JSON.parseObject(hiveJson, HiveJsonDto.class);

      if (hiveJsonDto.getWriteMode() != null) {
        if (hiveJsonDto.getWriteMode() == WriteMode.DO_NOTHING.getCode()) {
          hiveJsonDto.setWriteModeString(WriteMode.DO_NOTHING.getDescp());
        } else if (hiveJsonDto.getWriteMode() == WriteMode.ERROR.getCode()) {
          hiveJsonDto.setWriteModeString(WriteMode.ERROR.getDescp());
        }
      }

      if (hiveJsonDto.getFileType() != null) {
        if (hiveJsonDto.getFileType() == FileType.EXCEL2007.getCode()) {
          hiveJsonDto.setFileTypeString(FileType.EXCEL2007.getDescp());
        } else if (hiveJsonDto.getFileType() == FileType.CSV.getCode()) {
          hiveJsonDto.setFileTypeString(FileType.CSV.getDescp());
        } else if (hiveJsonDto.getFileType() == FileType.TEXT.getCode()) {
          hiveJsonDto.setFileTypeString(FileType.TEXT.getDescp());
        } else if (hiveJsonDto.getFileType() == FileType.ORC.getCode()) {
          hiveJsonDto.setFileTypeString(FileType.ORC.getDescp());
        }
      }

      // 方便前端统一从iotdbJsonDto中取值
      if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
        IotdbJsonDto iotdbJsonDto = new IotdbJsonDto();
        iotdbJsonDto.setSkipFileHeader(hiveJsonDto.getSkipFileHeader());
        iotdbJsonDto.setSkipLine(hiveJsonDto.getSkipLine());
        task.setIotdbJsonDto(iotdbJsonDto);
      }

      task.setHiveJsonDto(hiveJsonDto);
    }

    // 2、查询模板列字段
    List<ColumnInfo> srcColumnList = templateFieldService.getTemplateColumnListById(srcTableName);
    task.setSrcColumnList(srcColumnList);

    // 3、获取列映射关系
    Map<String, String> dstColumnMap = new HashMap<>();
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        dstColumnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }

    // 4、查询目标表列字段
    List<ColumnInfo> lastColumnList = new ArrayList<>();
    List<ColumnInfo> dstColumnList = new ArrayList<>();
    if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
      IotDbSessionUtils.initIotdbSessionPool(
          dstDatasource.getJdbcUrl(),
          dstDatasource.getJdbcUsername(),
          dstDatasource.getEncryptJdbcPassword());
      dstColumnList = IotDbSessionUtils.getColumnListByDevice(task.getDstTableName());
    } else {
      dstColumnList =
          baseDbService.getColumnList(jobDTO.getDstDatasourceId(), task.getDstTableName());
    }

    if (CollectionUtils.isNotEmpty(dstColumnList)) {
      for (ColumnInfo columnInfo : dstColumnList) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = dstColumnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    } else {
      for (ColumnInfo columnInfo : task.getSrcColumnList()) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = dstColumnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    }

    lastColumnList =
        taskService.sortBySrcColumn(srcColumnList, lastColumnList, task.getMappingColumn());
    task.setDstColumnList(lastColumnList);
    return task;
  }

  /** 更新任务 */
  @Override
  public String updateTask(TaskDTO oldTask, TaskDTO taskDTO) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(oldTask.getJobId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), oldTask.getJobId()));
    CheckParaUtils.checkObjectValidity(taskDTO.getModifyColumn(), logger, "是否修改列映射必填");

    // 记录日志
    BdpLogUtil.log4Update(taskDTO, oldTask);

    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
    if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
      HiveJsonDto hiveJsonDto = taskDTO.getHiveJsonDto();
      CheckParaUtils.checkObjectValidity(hiveJsonDto, logger, "hive配置必填");
      CheckParaUtils.checkStringValidity(hiveJsonDto.getFileName(), logger, "文件名必填");
      CheckParaUtils.checkObjectValidity(hiveJsonDto.getFileType(), logger, "文件类型必填");
      CheckParaUtils.checkStringValidity(hiveJsonDto.getFileSeparator(), logger, "分隔符必填");
      taskDTO.setHiveJson(JSONArray.toJSONString(hiveJsonDto));
    } else {
      IotdbJsonDto iotdbJsonDto = taskDTO.getIotdbJsonDto();
      CheckParaUtils.checkObjectValidity(iotdbJsonDto, logger, "iotdb配置必填");

      taskDTO.setIotdbJson(JSONArray.toJSONString(iotdbJsonDto));
    }

    // 判断列名称合法性
    if (!StringUtils.isEmpty(taskDTO.getMappingColumn())) {
      List<ColumnMappingDto> columnMappingDtos =
          JSON.parseArray(taskDTO.getMappingColumn(), ColumnMappingDto.class);
      this.checkColumnName(columnMappingDtos);
    }

    // 先更新列映射关系，再生成dataxjson
    taskService.updateByPrimaryKeySelective(taskDTO);
    return dataxJsonService.buildDataxJson(oldTask);
  }

  /**
   * 新增task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO insertTask(
      JobDTO jobDTO, TableMappingDto tableMappingDto, JobDatasource dstDatasource)
      throws Exception {
    String mappingColumnString =
        this.getMappingColumnString(
            tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), dstDatasource);
    TaskDTO taskDTO = new TaskDTO();
    taskDTO.setId(ComUtil.getId());
    taskDTO.setJobId(jobDTO.getId());
    taskDTO.setJobType(jobDTO.getJobType());
    taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
    taskDTO.setDstTableName(tableMappingDto.getDst_table());
    taskDTO.setMappingColumn(mappingColumnString);
    taskDTO.setSyncMode(jobDTO.getSyncMode());
    taskDTO.setFlag(Enable.ZERO.getCode());

    if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
      IotdbJsonDto iotdbJsonDto = new IotdbJsonDto();
      iotdbJsonDto.setEmptyLineProcess(Enable.ONE.getCode());
      iotdbJsonDto.setSkipFileHeader(Enable.ONE.getCode());
      iotdbJsonDto.setSkipLine(Enable.ONE.getCode());
      taskDTO.setIotdbJson(JSONArray.toJSONString(iotdbJsonDto));
    } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
      HiveJsonDto hiveJsonDto = new HiveJsonDto();
      hiveJsonDto.setDstTable(tableMappingDto.getDst_table());
      hiveJsonDto.setEmptyLineProcess(Enable.ONE.getCode());
      hiveJsonDto.setFileName(tableMappingDto.getDst_table());
      hiveJsonDto.setFileSeparator(Constants.COMMA);
      hiveJsonDto.setFileType(FileType.ORC.getCode());
      hiveJsonDto.setSkipFileHeader(Enable.ONE.getCode());
      hiveJsonDto.setSkipLine(Enable.ONE.getCode());
      hiveJsonDto.setWriteMode(WriteMode.DO_NOTHING.getCode());
      taskDTO.setHiveJson(JSONArray.toJSONString(hiveJsonDto));
    } else if (JdbcConstants.GREENPLUM.equals(dstDatasource.getDbType())
        || JdbcConstants.POSTGRESQL.equals(dstDatasource.getDbType())) {
      // 文件采集支持GP
      HiveJsonDto hiveJsonDto = new HiveJsonDto();
      hiveJsonDto.setSkipFileHeader(Enable.ONE.getCode());
      hiveJsonDto.setSkipLine(Enable.ONE.getCode());
      taskDTO.setHiveJson(JSONArray.toJSONString(hiveJsonDto));
    } else {
      throw new BusinessException(
          String.format(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dstDatasource.getDbType()));
    }

    taskService.insert(taskDTO);

    // 记录日志
    BdpLogUtil.log4Insert(taskDTO);
    logger.info("新增任务成功");

    return taskDTO;
  }

  /**
   * 更新task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO updateTask(
      JobDTO jobDTO, TableMappingDto tableMappingDto, JobDatasource dstDatasource)
      throws Exception {
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setSrcTableName(tableMappingDto.getSrc_table());
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO taskDTO = taskService.selectOne(query);

    // 目标表发生改变需要更新数据
    if (!taskDTO.getDstTableName().equals(tableMappingDto.getDst_table())) {
      String mappingColumnString =
          this.getMappingColumnString(
              tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), dstDatasource);
      taskDTO.setMappingColumn(mappingColumnString);

      if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
        IotdbJsonDto iotdbJsonDto = JSON.parseObject(taskDTO.getIotdbJson(), IotdbJsonDto.class);
        if (iotdbJsonDto == null) {
          iotdbJsonDto = new IotdbJsonDto();
        }
        iotdbJsonDto.setTimeseries(tableMappingDto.getDst_table());
        taskDTO.setIotdbJson(JSONArray.toJSONString(iotdbJsonDto));
      } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
        HiveJsonDto hiveJsonDto = JSON.parseObject(taskDTO.getHiveJson(), HiveJsonDto.class);
        if (hiveJsonDto == null) {
          hiveJsonDto = new HiveJsonDto();
        }
        hiveJsonDto.setDstTable(tableMappingDto.getDst_table());
        taskDTO.setHiveJson(JSONArray.toJSONString(hiveJsonDto));
      } else if (JdbcConstants.GREENPLUM.equals(dstDatasource.getDbType())
          || JdbcConstants.POSTGRESQL.equals(dstDatasource.getDbType())) {
        HiveJsonDto hiveJsonDto = new HiveJsonDto();
        hiveJsonDto.setSkipFileHeader(Enable.ONE.getCode());
        hiveJsonDto.setSkipLine(Enable.ONE.getCode());
        taskDTO.setHiveJson(JSONArray.toJSONString(hiveJsonDto));
      } else {
        throw new BusinessException(
            String.format(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dstDatasource.getDbType()));
      }

      taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
      taskDTO.setDstTableName(tableMappingDto.getDst_table());
      taskDTO.setFilterCondition(null);
      taskDTO.setSyncMode(jobDTO.getSyncMode());
      taskDTO.setModifyColumn(Enable.ZERO.getCode());
      taskDTO.setUpdateHistoryData(Enable.ONE.getCode());
      taskDTO.setSyncConditionField(SyncCondition.TIME.getCode());
      taskDTO.setFieldName(null);
      taskDTO.setSyncTimePoint(null);
      taskDTO.setSyncIdValue(null);
      taskService.updateByPrimaryKey(taskDTO);
      logger.info("修改任务成功");
    }

    // 删除上次任务保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
    TaskTableDetailDTO taskTableDetailDTO =
        TaskTableDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskTableDetailService.delete(taskTableDetailDTO);

    TaskDataDetailDTO taskDataDetailDTO =
        TaskDataDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskDataDetailService.delete(taskDataDetailDTO);
    return taskDTO;
  }

  /**
   * 删除任务
   *
   * @param jobDTO
   * @param srcTableList
   * @throws Exception
   */
  private void deleteTask(JobDTO jobDTO, List<String> srcTableList) throws Exception {
    // 根据jobId查询taskList，和selectedTable比较，selectedTable里边不存在的srcTable，就删除task，同时删除垃圾数据
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    List<String> srcTableMappingList =
        tableMappingDtoList.stream()
            .map(TableMappingDto::getSrc_table)
            .collect(Collectors.toList());
    List<String> collect =
        srcTableList.stream()
            .filter(s -> !srcTableMappingList.contains(s))
            .collect(Collectors.toList());
    collect.forEach(
        srcTableName -> {
          TaskDTO taskDtoQuery = new TaskDTO();
          taskDtoQuery.setJobId(jobDTO.getId());
          taskDtoQuery.setSrcTableName(srcTableName);
          taskDtoQuery.setFlag(Enable.ZERO.getCode());
          List<TaskDTO> taskList = taskService.selectList(taskDtoQuery);
          taskList.forEach(
              task -> {
                TaskTableDetailDTO taskTableDetailDTO = new TaskTableDetailDTO();
                taskTableDetailDTO.setTaskId(task.getId());
                taskTableDetailService.delete(taskTableDetailDTO);
                TaskDataDetailDTO taskDataDetailDTO = new TaskDataDetailDTO();
                taskDataDetailDTO.setTaskId(task.getId());
                taskDataDetailService.delete(taskDataDetailDTO);

                // 删除任务
                task.setFlag(1);
                taskService.updateByPrimaryKey(task);
              });
          //            taskService.delete(taskDtoQuery);
          logger.info("删除任务成功,srcTableName:{}", srcTableName);
        });
  }

  /**
   * 半结构化采集保存任务数据 一个job关联一个hdfs目录，可以选择目录下的多个模板，一个模板关联一张iotdb设备或者hive表
   *
   * @param jobDTO jobDTO
   * @throws Exception
   */
  @Override
  public void saveTasksByJobId(JobDTO jobDTO) throws Exception {
    List<String> srcTableList = taskService.getSrcTableListByJobId(jobDTO.getId());
    logger.info("新增或修改job时,taskList.size={}", srcTableList.size());

    // 根据jobId和srcTable查询task存不存在，存在就更新，不存在新增
    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);

    for (TableMappingDto tableMappingDto : tableMappingDtoList) {
      TaskDTO taskDTO = null;
      if (!srcTableList.contains(tableMappingDto.getSrc_table())) {
        taskDTO = insertTask(jobDTO, tableMappingDto, dstDatasource);
      } else {
        taskDTO = updateTask(jobDTO, tableMappingDto, dstDatasource);
      }

      // 保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
      taskService.saveTaskDetailData(
          tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), taskDTO, jobDTO);
    }

    // 删除任务
    deleteTask(jobDTO, srcTableList);
  }

  private Boolean syncPreHandle(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    boolean existDstTable =
        taskService.isExistTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
    logger.info("existDstTable:{}", existDstTable);
    taskService.saveTableLog(taskDTO, String.format("目标库存在目标表:%s", existDstTable));

    // 目标表不存在，直接新建表
    if (existDstTable) {
      // 目标已存在表的处理模式，预检查并报错拦截：0，覆盖目标表：1，保留目标表：2
      if (jobDTO.getProcessMode() == ProcessMode.PRECHECK_ERROR.getCode()) {
        JobDatasource dstDatasource =
            dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
        logger.info(
            DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
            dstDatasource.getJdbcUrl(),
            taskDTO.getDstTableName());
        throw new BusinessException(
            String.format(
                DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
                dstDatasource.getJdbcUrl(),
                taskDTO.getDstTableName()));
      } else if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()) {
        boolean isDel =
            baseDbService.dropTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("删除目标表:{},删除结果:{}", taskDTO.getDstTableName(), isDel);

        taskService.saveTableLog(
            taskDTO, String.format("删除目标表:%s,删除结果:%s", taskDTO.getDstTableName(), isDel));
      } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
        logger.info("已存在表的处理模式为保留目标表，不做处理");
        return false;
      } else if (jobDTO.getProcessMode() == ProcessMode.CLEAR_DST_TABLE_DATA.getCode()) {
        boolean isClear =
            baseDbService.clearTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("清空目标表数据:{},清空结果:{}", taskDTO.getDstTableName(), isClear);
        return false;
      } else {
        throw new BusinessException("不支持的已存在表的处理模式");
      }
    }

    return true;
  }

  /**
   * 创建目标表并回写建表sql语句
   *
   * @param jobDTO jobDTO
   * @param taskDTO taskDTO
   * @return 创建目标表结果
   * @throws Exception
   */
  @Override
  public Boolean createTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    logger.info("processMode:{}", jobDTO.getProcessMode());
    boolean createTable = syncPreHandle(jobDTO, taskDTO);

    if (!createTable) {
      return true;
    }

    // 列名称转换
    Map<String, String> columnMap = taskService.convertColumnName(taskDTO);

    // 1、在目标库创建表
    jobDTO.setTaskDTO(taskDTO);
    Map<String, String> map =
        baseDbService.createTable(
            jobDTO, taskDTO.getSrcTableName(), taskDTO.getDstTableName(), columnMap);

    // 2、回写创建表sql语句
    return taskService.updateSql(map, taskDTO);
  }

  /** 获取列映射关系 */
  private String getMappingColumnString(
      String srcTable, String dstTable, JobDatasource dstDatasource) throws Exception {
    Boolean isExistDstTable = taskService.isExistInDstDatabase(dstDatasource.getId(), dstTable);
    List<ColumnMappingDto> columnMappingDtoList;
    if (isExistDstTable) {
      // 目标数据库存在目标表
      // 获取源表列名称以及类型
      List<ColumnInfo> srcColumnList = templateFieldService.getTemplateColumnListById(srcTable);
      // 获取目标表列名称以及类型
      List<ColumnInfo> dstColumnList = new ArrayList<>();
      if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
        IotDbSessionUtils.initIotdbSessionPool(
            dstDatasource.getJdbcUrl(),
            dstDatasource.getJdbcUsername(),
            dstDatasource.getEncryptJdbcPassword());
        dstColumnList = IotDbSessionUtils.getColumnListByDevice(dstTable);
      } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())
          || JdbcConstants.GREENPLUM.equals(dstDatasource.getDbType())
          || JdbcConstants.POSTGRESQL.equals(dstDatasource.getDbType())) {
        dstColumnList = baseDbService.getColumnList(dstDatasource.getId(), dstTable);
      } else {
        logger.debug(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dstDatasource.getDbType());
      }

      columnMappingDtoList = taskService.getMergeColumnList(srcColumnList, dstColumnList);
    } else {
      // 目标数据库不存在目标表 获取源表列名称以及类型
      List<ColumnInfo> mergeColumnList = templateFieldService.getTemplateColumnListById(srcTable);
      columnMappingDtoList =
          taskService.getColumnMappingDtos(mergeColumnList, dstDatasource.getDbType());
    }

    logger.info("列映射关系为:{}", JSONArray.toJSONString(columnMappingDtoList));
    return JSONArray.toJSONString(columnMappingDtoList);
  }

  /** 检查列名合法性 */
  private void checkColumnName(List<ColumnMappingDto> list) {
    for (ColumnMappingDto columnMappingDto : list) {
      if (!tableService.isValidColumnName(columnMappingDto.getDstColumnName())) {
        throw new BusinessException(
            String.format(
                DcsErrorCode.COLUMN_NAME_CHECK.getDesc(), columnMappingDto.getDstColumnName()));
      }
    }
  }

  @Override
  public Integer getType() {
    return JobType.SEMISTRUCTURE.getCode();
  }
}
